package com.jacquesh;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;



public class FlinkDemo0010 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = environment
                .socketTextStream("192.168.88.180", 8888);
        IterativeStream<Long> iterate = stream
                .map(new RichMapFunction<String, Long>() {
                    @Override
                    public Long map(String s) {
                        return Long.parseLong(s);
                    }
                }).iterate();

        DataStream<Long> feedback = iterate
                .map(new RichMapFunction<Long, Long>() {


                    @Override
                    public Long map(Long aLong) throws Exception {

                        return aLong > 2 ? aLong : (aLong + 1);
                    }
                }).filter(new FilterFunction<Long>() {
                    @Override
                    public boolean filter(Long s) throws Exception {
                        if (s > 2) {
                            return false;
                        } else {
                            return true;
                        }
                    }
                });
        iterate.closeWith(feedback).print("feedback");
        SingleOutputStreamOperator<Long> result = iterate.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return value >= 2;
            }
        });
        result.print("result:");
        environment.execute("iterate demo");
    }
}
